<html>
 <head>
  <meta charset="UTF-8">
 </head>
 <body>
  <h1 data-lake-id="zANCC" id="zANCC"><span data-lake-id="u591dacaf" id="u591dacaf">典型回答</span></h1>
  <p data-lake-id="u5f51f07f" id="u5f51f07f"><br></p>
  <p data-lake-id="u65a15709" id="u65a15709"><span data-lake-id="udfc6f71f" id="udfc6f71f">什么是消费端限流，这是一种保护消费者的手段，假如说，现在是业务高峰期了，消息有大量堆积，导致MQ消费者需要不断地进行消息消费，很容易被打挂，甚至重启之后还是会被大量消息涌入，继续被打挂。</span></p>
  <p data-lake-id="uc136a694" id="uc136a694"><span data-lake-id="ub3d6ea38" id="ub3d6ea38">​</span><br></p>
  <p data-lake-id="u5d04b088" id="u5d04b088"><span data-lake-id="ue48c063f" id="ue48c063f">为了解决这个问题，RabbitMQ提供了basicQos的方式来实现消费端限流。我们可以在消费者端指定最大的未确认消息数，当达到这个限制时，RabbitMQ将不再推送新的消息给消费者，直到有一些消息得到确认。</span></p>
  <p data-lake-id="ud45ee921" id="ud45ee921"><span data-lake-id="u94bf7e9f" id="u94bf7e9f">​</span><br></p>
  <p data-lake-id="uea5667f9" id="uea5667f9"><span data-lake-id="u2493cb8e" id="u2493cb8e">想要实现这个功能，首先需要把自动提交关闭。</span></p>
  <p data-lake-id="u479353de" id="u479353de"><span data-lake-id="u867bd3fb" id="u867bd3fb">​</span><br></p>
  <pre lang="java"><code>
channel.basicConsume(queueName, false, consumer);
</code></pre>
  <p data-lake-id="u8f26dc2f" id="u8f26dc2f"><span data-lake-id="u32b731cb" id="u32b731cb">​</span><br></p>
  <p data-lake-id="u866a3104" id="u866a3104"><span data-lake-id="ud4034d55" id="ud4034d55">接着进行限流配置：</span></p>
  <pre lang="java"><code>
/**
  * 限流设置:  
	*	prefetchSize：每条消息大小的设置，0是无限制
  * prefetchCount:标识每次推送多少条消息
  * global:false标识channel级别的  true:标识消费者级别的
  */
 channel.basicQos(0,10,false);
</code></pre>
  <p data-lake-id="uf73649f0" id="uf73649f0"><br></p>
  <p data-lake-id="uc7e68c36" id="uc7e68c36"><span data-lake-id="u028e8565" id="u028e8565">如以上配置，可以实现消费者在处理完一条消息后，才会获取下一条消息。</span></p>
  <p data-lake-id="uda8853b8" id="uda8853b8"><span data-lake-id="ub13091b6" id="ub13091b6">​</span><br></p>
  <p data-lake-id="u4200cbd9" id="u4200cbd9"><span data-lake-id="ucb771b9d" id="ucb771b9d">然后再在消费者处理完一条消息之后，手动发送确认消息给到RabbitMQ，这样就可以拉取下一条消息了：</span></p>
  <p data-lake-id="uae4295d7" id="uae4295d7"><span data-lake-id="u2a94c97f" id="u2a94c97f">​</span><br></p>
  <pre lang="java"><code>
channel.basicAck(deliveryTag, false); // 发送确认
</code></pre>
  <p data-lake-id="uf4b93eb0" id="uf4b93eb0"><br></p>
  <p data-lake-id="u2fd58a75" id="u2fd58a75"><span data-lake-id="u6ce95334" id="u6ce95334">完整代码如下：</span></p>
  <p data-lake-id="u51b9e648" id="u51b9e648"><span data-lake-id="ua2af7410" id="ua2af7410">​</span><br></p>
  <pre lang="java"><code>
import com.rabbitmq.client.*;

public class ConsumerWithFlowControl {
    private static final String QUEUE_NAME = "my_queue";
    private static final String HOST = "localhost";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            // 设置消费者限流，每次只获取一条消息
            int prefetchCount = 1;
            channel.basicQos(prefetchCount);

            // 创建消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println("Received: " + message);

                    // 模拟消息处理耗时
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }

                    // 发送消息确认
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };

            // 指定队列，并关闭自动确认
            channel.basicConsume(QUEUE_NAME, false, consumer);

           
        }
    }
}

</code></pre>
 </body>
</html>